package com.dtruth.dataflow.bolt;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;

import com.dtruth.dataflow.dao.RedisDAO;
import com.dtruth.dataflow.util.Env;
import com.dtruth.dataflow.util.HtmlRegexpUtil;
import com.dtruth.dataflow.util.MyLoggerFactory;
import com.dtruth.dataflow.util.StringUtil;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class DataNormalBolt extends BaseBasicBolt {

	private static Logger log = MyLoggerFactory.getLogger(DataNormalBolt.class);
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	RedisDAO rd = null;

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		String host = Env.getInstance().getProperty("redis.dataconfig.host")
				.trim();
		int port = Integer.valueOf(Env.getInstance()
				.getProperty("redis.dataconfig.port").trim());
		int insNum = Integer.valueOf(Env.getInstance()
				.getProperty("redis.dataconfig.db").trim());
		rd = new RedisDAO(host, port, insNum);
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		Object domain = map.get("domain");
		try {
			if (domain != null) {
				Object pdStr = map.get("pubdate");
				if (pdStr != null && !pdStr.toString().equals("null")) {
					long pubdate = pubdate(domain.toString(),
							map.get("pubdate"));
					if (pubdate != -1) {
						map.put("pubdate", pubdate);
					} else {
						log.warn(map.get("url") + ":pubdate="
								+ map.get("pubdate"));
						map.remove("pubdate");
					}
				} else {
					log.warn(map.get("url") + ":pubdate is null");
					map.put("pubdate",
							System.currentTimeMillis() - 1000 * 60 * 60);
				}
			}
			String category = category(map);
			if (category.length() > 0 && !category.equals(""))
				map.put("category", category);

			Object summary = map.get("summary");
			if (summary != null) {
				summary = HtmlRegexpUtil.getValidContent(summary.toString());
				if (summary.toString().length() > 255)
					map.put("summary", "");
				else
					map.put("summary", summary);
			}
			Object origin = map.get("origin");// 有的来源配置的时候写了链接
			if (origin != null) {
				String originStr = HtmlRegexpUtil.getValidContent(origin
						.toString());
				if (!StringUtil.containsCNStr(origin.toString())
						&& origin.toString().length() > 20) {
					log.warn("wrong origin:" + origin + "(" + map.get("url")
							+ ")");
					map.put("origin", "");
				} else {
					if (StringUtil.containsDot(originStr))
						log.warn("originStr contains dot:" + origin + "("
								+ map.get("url") + ")");
					map.put("origin", originStr);
				}
			}

			Object author = map.get("author");
			if (author != null && author.toString().length() > 32) {
				log.warn("author is too long:" + author + "(" + map.get("url")
						+ ")");
				map.remove("author");
			}
			log.debug("emit:" + map.get("url"));
			collector.emit(new Values(map));
		} catch (Exception e) {
			log.warn(ExceptionUtils.getStackTrace(e) + ":" + map.get("url"));

		}

	}

	private long pubdate(String domain, Object pubdate) {
		if (domain == null)
			return -1;
		Set<String> dfs = rd.smembers("domain:" + domain);
		if (dfs != null && dfs.size() > 0) {
			for (String df : dfs) {
				SimpleDateFormat sdf = new SimpleDateFormat(df);
				try {
					Date d = sdf.parse(pubdate.toString().trim());
					return d.getTime();
				} catch (ParseException e) {
					continue;
				}
			}
		}
		dfs.clear();
		dfs.add("yyyy-MM-dd HH:mm:ss");
		dfs.add("yyyy-MM-dd");
		dfs.add("yyyy-MM-dd HH:mm");
		for (String df : dfs) {
			SimpleDateFormat sdf = new SimpleDateFormat(df);
			try {
				Date d = sdf.parse(pubdate.toString().trim());
				return d.getTime();
			} catch (ParseException e) {
				continue;
			}
		}
		return -1;
	}

	/**
	 * 重新定义分类
	 * 
	 * @param map
	 * @return
	 */
	private String category(Map<String, Object> map) {
		StringBuffer category = new StringBuffer();
		Object obj = map.get("category1");
		if (obj != null) {
			map.remove("category1");
			category.append(obj.toString() + ",");
			obj = map.get("category2");
			if (obj != null) {
				map.remove("category2");
				category.append(obj.toString() + ",");
				obj = map.get("category3");
				if (obj != null) {
					map.remove("category3");
					category.append(obj.toString() + ",");
				}
			}
		}
		return category.toString();
	}
}